package com.fzk.server;

import com.alibaba.fastjson2.JSON;
import com.fzk.core.FileMeta;
import com.fzk.core.FileMetaStore;
import com.fzk.core.FileStoreService;
import com.fzk.log.Logger;
import com.fzk.utils.MyFileUtil;
import com.fzk.vo.CheckAck;
import com.fzk.vo.WsMsg;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.*;

import java.io.IOException;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.time.Instant;
import java.util.Map;
import java.util.Set;

/**
 * @author zhike.feng
 * @datetime 2023-06-22 11:01:35
 */
public class WebSocketFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
    private final FileStoreService fileService;
    private Path storagePath;// 分片上传的文件路径
    private FileChannel fileChannel;// 分片上传的文件通道
    private FileMetaStore.FileMetaWrapper metaWrapper;// 分片上传的文件元信息
    private long writePos;

    public WebSocketFrameHandler(FileStoreService fileService) {
        this.fileService = fileService;
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        clear();
        ctx.fireChannelInactive();// 继续传递该事件
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws IOException {
        // ping and pong frames already handled
        if (frame instanceof TextWebSocketFrame) {
            // Send the uppercase string back.
            String msg = ((TextWebSocketFrame) frame).text();
            Logger.debug("ws消息: " + msg);
            WsMsg wsmsg = JSON.parseObject(msg, WsMsg.class);
            switch (wsmsg.type) {
                case WsMsg.Type_Checkpoint -> {
                    String filename = String.valueOf(wsmsg.headers.getOrDefault("filename", ""));
                    String bucket = String.valueOf(wsmsg.headers.getOrDefault("bucket", ""));
                    String relativePath = String.valueOf(wsmsg.headers.getOrDefault("relativePath", ""));
                    long size = Long.parseLong(String.valueOf(wsmsg.headers.getOrDefault("size", "0")));
                    if (filename.length() == 0 || bucket.length() == 0 || relativePath.length() == 0 || size <= 0) {
                        sendText(ctx, new WsMsg(WsMsg.Type_Err, Map.of("code", 400, "msg", "参数错误"), null));
                        ctx.channel().close();
                        return;
                    }
                    CheckAck checkAck = fileService.fileSliceCheck(bucket, relativePath, filename);
                    sendText(ctx, new WsMsg(WsMsg.Type_Checkpoint, Map.of("key", checkAck.key, "ack", checkAck.ack), checkAck));
                }
                case WsMsg.Type_Ack -> throw new RuntimeException("不支持该消息类型");

                case WsMsg.Type_FileUploadStart -> {
                    String filename = String.valueOf(wsmsg.headers.getOrDefault("filename", ""));
                    String bucket = String.valueOf(wsmsg.headers.getOrDefault("bucket", ""));
                    String relativePath = String.valueOf(wsmsg.headers.getOrDefault("relativePath", ""));
                    long size = Long.parseLong(String.valueOf(wsmsg.headers.getOrDefault("size", "0")));
                    long position = Long.parseLong(String.valueOf(wsmsg.headers.getOrDefault("position", "0")));
                    if (filename.length() == 0 || bucket.length() == 0 || relativePath.length() == 0 || size <= 0) {
                        WsMsg errMsg = new WsMsg(WsMsg.Type_Err, Map.of("code", "400", "msg", "参数错误"), null);
                        ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(errMsg)));
                        ctx.channel().close();
                        return;
                    }
                    Logger.debug(String.format("开始分片上传文件: buket: %s, relativePath: %s, filename: %s, size: %s", bucket, relativePath, filename, size));

                    clear();// 先清空可能存在的旧数据
                    this.storagePath = fileService.generateStoragePath(Path.of(bucket, relativePath, filename).toString());
                    // 确保文件已创建
                    MyFileUtil.ensureDirsAndFile(storagePath);
                    // 打开文件通道
                    this.fileChannel = FileChannel.open(storagePath, Set.of(StandardOpenOption.WRITE));
                    // 写入元信息和索引
                    this.metaWrapper = fileService.putMeta(null, storagePath, FileStoreService.generateKey(storagePath, fileService.dataDir), bucket, FileMeta.Uploading, size);
                    this.writePos = position;
                }
                case WsMsg.Type_FileUploadFinish -> {
                    Logger.debug(String.format("文件分片上传完成, storagePath: %s, meta: %s", storagePath, metaWrapper));
                    checkNotNull();
                    // 落盘并重置
                    clear();
                }
                default -> throw new RuntimeException("未知的消息类型: " + wsmsg.type);
            }
        } else if (frame instanceof BinaryWebSocketFrame || frame instanceof ContinuationWebSocketFrame) {
            // 字节桢是文件分片上传内容
            // 检查
            checkNotNull();
            ByteBuf buf = frame.content();
            while (buf.isReadable()) {
                writePos += buf.readBytes(fileChannel, writePos, buf.readableBytes());
            }
            // 因为桢可能因太长而分段, 所以只有最后一段需要发送消息
            if (frame.isFinalFragment()) {
                sendText(ctx, new WsMsg(WsMsg.Type_Ack, Map.of("key", metaWrapper.meta.getKey(), "ack", writePos), null));
            }
        } else {
            // 注意：像ping/pong/close等消息桢早被处理了不会走到这里
            String message = "unsupported frame type: " + frame.getClass().getName();
            throw new UnsupportedOperationException(message);
        }
    }

    private void sendText(ChannelHandlerContext ctx, WsMsg msg) {
        ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(msg)));
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        // ws协议握手成功事件
        if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete handshake) {
            Logger.debug(String.format("ws 握手成功: uri: %s, \nheaders: %s", handshake.requestUri(), handshake.requestHeaders()));
        }
    }

    public void checkNotNull() {
        if (storagePath == null || fileChannel == null || metaWrapper == null) {
            throw new RuntimeException(String.format("存在null, storagePath: %s, fileChannel: %s, metaWrapper: %s", storagePath, fileService, metaWrapper));
        }
        if (!fileChannel.isOpen()) {
            throw new RuntimeException("文件通道已经关闭?");
        }
        if (writePos < 0) {
            throw new RuntimeException("writePos<0? writePos=" + writePos);
        }
    }

    public void clear() throws IOException {
        if (fileChannel != null && fileChannel.isOpen()) {
            fileChannel.force(true);// 落盘
            fileChannel.close();
            // 更新文件元信息, 落盘文件
            if (writePos >= metaWrapper.meta.getSize() && metaWrapper.meta.getUploadStatus() == FileMeta.Uploading) {
                fileService.sliceUploadFinish(storagePath, metaWrapper, writePos);
                FileMeta meta = metaWrapper.meta;
                long costMills = meta.getUploadEndTimeStamp() - meta.getUploadStartTimeStamp();
                Logger.info(String.format("文件ws上传结束, filename: %s, start: %s, end: %s, cost: %d.%d 秒",
                        meta.getFilename(),
                        Instant.ofEpochMilli(meta.getUploadStartTimeStamp()),
                        Instant.ofEpochMilli(meta.getUploadEndTimeStamp()),
                        costMills / 1000, costMills % 1000));
            }
        }
        fileChannel = null;
        storagePath = null;
        metaWrapper = null;
        writePos = 0L;
    }

    @Override// 错误拦截
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        Logger.error(cause.getMessage());
        if (ctx.channel().isActive()) {
            WsMsg errMsg = new WsMsg(WsMsg.Type_Err, Map.of("code", "500", "msg", "内部错误: " + cause.getMessage()), null);
            ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(errMsg)));
            ctx.channel().close();
        }
    }
}
